package com.jml.顺序消息;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class ProducerOrder {
    // 创建我们的分组
    DefaultMQProducer producer = new DefaultMQProducer("mayiktGroup");

    public void sendMsg() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        // 设置我们的NamesrvAddr
        producer.setNamesrvAddr("192.168.212.169:9876");
        producer.start();

        Long orderId = System.currentTimeMillis();
        String insertSql = getSqlMsg("insert", orderId);
        String updateSql = getSqlMsg("update", orderId);
        String deleteSql = getSqlMsg("delete", orderId);
        Message insertMsg = new Message("mtTopic", insertSql.getBytes());
        Message updateMsg = new Message("mtTopic", updateSql.getBytes());
        Message deleteMsg = new Message("mtTopic", deleteSql.getBytes());

        producer.send(insertMsg, new MessageQueueSelector() {
            /**
             * @param mqs:集群队列总数
             * @param msg：消息内容
             */
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // 设置该消息永远存放到第0个队列中，mqs.get(0)。
                Long orderId = (Long) arg;
                long index = orderId % mqs.size();
                //表示该消息固定投放到某一个队列中
                return mqs.get((int) index);
            }
        }, orderId);
        producer.send(updateMsg, new MessageQueueSelector() {
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // 设置该消息永远存放到第0个队列中，mqs.get(0)。
                Long orderId = (Long) arg;
                long index = orderId % mqs.size();
                return mqs.get((int) index);
            }
        }, orderId);
        producer.send(deleteMsg, new MessageQueueSelector() {
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // 设置该消息永远存放到第0个队列中。
                Long orderId = (Long) arg;
                long index = orderId % mqs.size();
                return mqs.get((int) index);
            }
        }, orderId);


    }

    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        new ProducerOrder().sendMsg();
    }

    public String getSqlMsg(String type, Long orderId) {
        JSONObject dataObject = new JSONObject();
        dataObject.put("type", type);
        dataObject.put("orderId", orderId);
        return dataObject.toJSONString();
    }
}

